/*
 * ClientConnection.java
 *
 * Created on August 21, 2007, 1:39 PM
 *
 * To change this template, choose Tools | Template Manager
 * and open the template in the editor.
 */

package org.xeerkat.xmpp;

import java.io.FilterReader;
import java.io.FilterWriter;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.Reader;
import java.io.StringWriter;
import java.io.Writer;
import java.net.Socket;
import java.util.Hashtable;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.naming.Context;
import javax.naming.NamingException;
import javax.naming.directory.Attribute;
import javax.naming.directory.Attributes;
import javax.naming.directory.DirContext;
import javax.naming.directory.InitialDirContext;
import javax.net.SocketFactory;
import javax.net.ssl.HandshakeCompletedEvent;
import javax.net.ssl.HandshakeCompletedListener;
import javax.net.ssl.SSLSocket;
import javax.net.ssl.SSLSocketFactory;
import org.infoset.xml.Characters;
import org.infoset.xml.DestinationSource;
import org.infoset.xml.Document;
import org.infoset.xml.Element;
import org.infoset.xml.Item;
import org.infoset.xml.ItemConstructor;
import org.infoset.xml.ItemDestination;
import org.infoset.xml.Name;
import org.infoset.xml.Named;
import org.infoset.xml.XMLException;
import org.infoset.xml.jaxp.StreamItemSource;
import org.infoset.xml.util.DocumentDestination;
import org.infoset.xml.util.XMLWriter;

/**
 *
 * @author alex
 */
public class ClientConnection
{
   static final ThreadGroup threadGroup = new ThreadGroup("xmpp.client");
   
   public static ThreadGroup getThreadGroup() {
      return threadGroup;
   }

   static Logger LOG = Logger.getLogger(ClientConnection.class.getName());
   public static ClientConnection lookupByDomain(String domain)
      throws NamingException
   {
      Hashtable env = new Hashtable();
      env.put(Context.INITIAL_CONTEXT_FACTORY, "com.sun.jndi.dns.DnsContextFactory");
      DirContext ictx = new InitialDirContext(env);
      try {
         Attributes dnsLookup = ictx.getAttributes("_xmpp-client._tcp." + domain, new String[]{"SRV"});
         Attribute srvAtt = dnsLookup.get("SRV");
         if (srvAtt!=null) {
            String srvRecord = (String)srvAtt.get();
            String [] srvRecordEntries = srvRecord.split(" ");
            int port = Integer.parseInt(srvRecordEntries[srvRecordEntries.length-2]);
            String host = srvRecordEntries[srvRecordEntries.length-1];
            return new ClientConnection(host,port);
         } else {
            return new ClientConnection(domain);
         }
      } catch (javax.naming.NameNotFoundException ex) {
         return new ClientConnection(domain);
      }
   }
   
   long keepAliveSentAt = 0;
   long reconnectWait = 10000;
   long idleTimeout = 2*60*1000;
   boolean autoReconnect;
   
   String host;
   int port;
   boolean portSecure;
   boolean secure;
   boolean makeSecure;
   
   Socket socket;
   Writer serverW;
   Reader serverR;
   Writer outgoingLog;
   Writer incomingLog;
   StanzaReader incoming;
   StanzaWriter outgoing;
   
   ID id;
   ID bindId;
   
   AtomicBoolean receiverRunning;
   Thread receiverThread;
   
   AtomicBoolean keepAliveRunning;
   Thread keepAliveThread;
   
   Map<Name,String> properties;
   
   ItemConstructor constructor;
   boolean connecting = false;
   boolean connected = false;
   
   /**
    * Creates a new instance of ClientConnection
    */
   public ClientConnection(String host)
   {
      this(host,5222);
   }
   
   public ClientConnection(String host,int port)
   {
      this(host,port,false);
   }
   
   public ClientConnection(String host,int port,boolean secure)
   {
      this.host = host;
      this.port = port;
      this.portSecure = secure;
      this.secure = secure;
      this.socket = null;
      this.properties = new TreeMap<Name,String>();
      this.makeSecure = true;
      this.autoReconnect = false;
      this.keepAliveRunning = new AtomicBoolean(false);
      this.outgoingLog = null;
      this.incomingLog = null;
   }
   
   public long getReconnectWait() {
      return reconnectWait;
   }
   
   public void setReconnectWait(long wait) {
      reconnectWait = wait;
   }
   
   public void setAutoReconnect(boolean flag)
   {
      this.autoReconnect = flag;
   }
   
   public void setIdleTimeout(long wait) {
      idleTimeout = wait;
   }
   
   public long getIdleTimeout() {
      return idleTimeout;
   }
   
   public boolean isAutoReconnecting() {
      return this.autoReconnect;
   }
   
   public void setRequireSecure(boolean flag)
   {
      this.makeSecure = flag;
   }
   
   public boolean isSecure() {
      return secure;
   }
   
   public ID getID() {
      return id;
   }
   
   public ID getBindID() {
      return bindId;
   }

   public ItemConstructor getItemConstructor() {
      return constructor;
   }

   public void setOutgoingLog(Writer log) {
      this.outgoingLog = log;
   }

   public void setIncomingLog(Writer log) {
      this.incomingLog = log;
   }

   private String toString(Document doc) {
      StringWriter w = new StringWriter();
      try {
         XMLWriter.writeDocument(doc, w);
      } catch (Exception ex) {
      }
      return w.toString();
   }
   
   private String toString(Element e) {
      StringWriter w = new StringWriter();
      try {
         XMLWriter.writeElement(e, w);
      } catch (Exception ex) {
      }
      return w.toString();
   }

   public void connect(ID id) 
      throws IOException,XMLException
   {
      try {
         
         connecting = true;
         connected = false;

         this.id = id;

         properties.put(StanzaReader.TO_PROPERTY,id.getDomain());
         boolean isFine = LOG.isLoggable(Level.FINE);

         if (isFine) {
            LOG.fine(id+" connecting to "+host+":"+port);
         }

         if (portSecure) {
            SocketFactory sslFactory = SSLSocketFactory.getDefault();
            socket = (SSLSocket)sslFactory.createSocket(host,port);
            secure = true;
            LOG.fine("Secure socket established.");
         } else {
            socket = new Socket(host,port);
            LOG.fine("Socket established.");
         }


         OutputStream serverOS = socket.getOutputStream();
         serverW = new OutputStreamWriter(serverOS,"UTF-8");
         InputStream serverIS = socket.getInputStream();
         serverR = new InputStreamReader(serverIS,"UTF-8");

         outgoing = new OrderedStanzaWriter(LOG,serverW);
         constructor = outgoing.getConstructor();
         outgoing.start(properties);

         incoming = new StanzaReader(LOG,new StreamItemSource(null,serverR,true));
         incoming.start();

         boolean starttls = false;
         if ("1.0".equals(incoming.getProperties().get(StanzaReader.VERSION_PROPERTY))) {
            DocumentDestination dest = new DocumentDestination();
            if (incoming.next(dest)) {
               Document doc = dest.getDocument();
               Element top = doc.getDocumentElement();
               Name name = top.getName();
               if (name.equals(Protocol.FEATURES)) {
                  starttls = top.getFirstElementNamed(Protocol.STARTTLS)!=null;
               }
            } else {
               closeSocket(false);
               throw new XMLException("Features could not be received as stream was closed.");
            }
         } else {
            starttls = true;
         }
         if (starttls || makeSecure) {
            LOG.fine("Starting TLS...");
            ItemDestination stanza = outgoing.next();
            stanza.send(constructor.createElement(Protocol.STARTTLS));
            stanza.send(constructor.createElementEnd(Protocol.STARTTLS));

            DocumentDestination proceed = new DocumentDestination();
            if (!incoming.next(proceed)) {
               closeSocket(false);
               throw new XMLException("Failed to receive proceed stanza.");
            }
            if (!proceed.getDocument().getDocumentElement().getName().equals(Protocol.STARTTLS_PROCEED)) {
               closeSocket(true);
               throw new XMLException("TLS start failed: "+toString(proceed.getDocument()));
            }
            LOG.fine("Proceeding with TLS ...");

            SSLSocketFactory sslFactory = (SSLSocketFactory)SSLSocketFactory.getDefault();
            SSLSocket sslSocket = (SSLSocket)sslFactory.createSocket(socket,socket.getInetAddress().getHostName(), socket.getPort(), false);
            //sslSocket.setSoTimeout(0);
            sslSocket.setSoTimeout((int)(idleTimeout/2));
            sslSocket.setKeepAlive(true);
            final CyclicBarrier wait = new CyclicBarrier(2);
            sslSocket.addHandshakeCompletedListener(new HandshakeCompletedListener() {
               public void handshakeCompleted(HandshakeCompletedEvent event) {
                  try {
                     wait.await();
                  } catch (Exception ex) {
                     ex.printStackTrace();
                  }
               }
            });
            sslSocket.startHandshake();
            try {
               wait.await();
            } catch (Exception ex) {
               closeSocket(false);
               throw new XMLException("Negotiation handshake was interrupted.",ex);
            }
            secure = true;
            socket = sslSocket;

            serverW = new OutputStreamWriter(socket.getOutputStream(),"UTF-8");
            serverR = new InputStreamReader(socket.getInputStream(),"UTF-8");

            LOG.fine("Connected over TLS...");

            outgoing = new OrderedStanzaWriter(LOG,serverW);

            constructor = outgoing.getConstructor();

            outgoing.start(properties);

            incoming = new StanzaReader(LOG,new StreamItemSource(null,serverR,true));
            incoming.start();

            final Set<String> mechanisms = new TreeSet<String>();
            if ("1.0".equals(incoming.getProperties().get(StanzaReader.VERSION_PROPERTY))) {
               incoming.next(new ItemDestination() {
                  StringBuilder builder = null;
                  public void send(Item item)
                  {
                     switch (item.getType()) {
                        case ElementItem:
                           if (((Named)item).getName().equals(Protocol.MECHANISM)) {
                              builder = new StringBuilder();
                           }
                           break;
                        case ElementEndItem:
                           if (((Named)item).getName().equals(Protocol.MECHANISM)) {
                              mechanisms.add(builder.toString());
                              builder = null;
                           }
                           break;
                        case CharactersItem:
                           if (builder!=null) {
                              builder.append(((Characters)item).getText());
                           }
                     }
                  }
               });
            } else {
               mechanisms.add("PLAIN");
            }
            if (isFine) {
               LOG.fine("Mechanisms: "+mechanisms);
            }

            if (id.getPassword()!=null) {
               if (!mechanisms.contains("PLAIN")) {
                  StringBuilder builder = new StringBuilder();
                  for (String name : mechanisms) {
                     builder.append(" ");
                     builder.append(name);
                  }
                  closeSocket(true);
                  throw new XMLException("PLAIN authentication over TLS is not supported: "+builder.toString());
               }
               LOG.fine("Authenticating...");
               //LOG.fine("credentials: "+id.getName()+" "+id.getPassword());

               ItemDestination authStanza = outgoing.next();
               Element auth = constructor.createElement(Protocol.AUTH);
               auth.setAttributeValue("mechanism", "PLAIN");
               authStanza.send(auth);
               authStanza.send(constructor.createCharacters(Base64Coder.encode(id.toXMPPId()+"\0"+id.getName()+"\0"+id.getPassword())));
               authStanza.send(constructor.createElementEnd(Protocol.AUTH));

               DocumentDestination authResponse = new DocumentDestination();
               incoming.next(authResponse);
               if (authResponse.getDocument().getDocumentElement().getName().equals(Protocol.FAILURE)) {
                  throw new XMLException("Authentication failed "+toString(authResponse.getDocument()));
               }
               /*
               incoming.next(new ItemDestination() {
                  public void send(Item item) 
                     throws XMLException
                  {
                     switch (item.getType()) {
                        case ElementItem:
                           if (((Named)item).getName().equals(Protocol.FAILURE)) {
                              throw new XMLException("Authentication failed");
                           }
                     }
                  }
               });*/

               LOG.fine("Re-initializing stream...");

               Writer connectionWriter = serverW;
               if (outgoingLog!=null) {
                  connectionWriter = new FilterWriter(serverW) {
                     public void write(char [] buffer, int offset, int len)
                        throws IOException
                     {
                        outgoingLog.write(buffer, offset, len);
                        outgoingLog.flush();
                        super.write(buffer,offset,len);
                     }
                     public void write(int c)
                        throws IOException
                     {
                        outgoingLog.write(c);
                        outgoingLog.flush();
                        super.write(c);
                     }
                     public void write(String str, int offset, int len)
                        throws IOException
                     {
                        outgoingLog.write(str, offset, len);
                        outgoingLog.flush();
                        super.write(str,offset,len);
                     }
                  };
               }
               Reader connectionReader = serverR;
               if (incomingLog!=null) {
                  connectionReader = new FilterReader(serverR) {
                     public int read(char [] buffer, int offset, int len)
                        throws IOException
                     {
                        int size = super.read(buffer,offset,len);
                        if (size>0) {
                           incomingLog.write(buffer, offset, size);
                           incomingLog.flush();
                        }
                        return size;
                     }
                     public int read()
                        throws IOException
                     {
                        int c = super.read();
                        if (c>0) {
                           outgoingLog.write(c);
                           outgoingLog.flush();
                        }
                        return c;
                     }
                  };
               }
               outgoing = new OrderedStanzaWriter(LOG,connectionWriter);

               constructor = outgoing.getConstructor();

               outgoing.start(properties);

               incoming = new StanzaReader(LOG,new StreamItemSource(null,connectionReader,true));
               incoming.start();

               LOG.fine("Proceeding...");

               if ("1.0".equals(incoming.getProperties().get(StanzaReader.VERSION_PROPERTY))) {
                  incoming.next(null);
               }
            }

         }

         ItemDestination bindStanza = outgoing.next();
         Element bindIQ = constructor.createElement(Protocol.IQ);
         bindIQ.setAttributeValue("type","set");
         bindIQ.setAttributeValue("id", "bind");
         bindStanza.send(bindIQ);
         bindStanza.send(constructor.createElement(Protocol.BIND));
         if (id.getResource()!=null) {
            if (isFine) {
               LOG.fine("Binding to "+id.getResource());
            }
            bindStanza.send(constructor.createElement(Protocol.RESOURCE));
            bindStanza.send(constructor.createCharacters(id.getResource()));
            bindStanza.send(constructor.createElementEnd(Protocol.RESOURCE));
         } else {
            LOG.fine("Binding to server assigned resource.");
         }
         bindStanza.send(constructor.createElementEnd(Protocol.BIND));
         bindStanza.send(constructor.createElementEnd(Protocol.IQ));

         DocumentDestination iqDest = new DocumentDestination();
         if (incoming.next(iqDest)) {
            Element top = iqDest.getDocument().getDocumentElement();
            if (top.getName().equals(Protocol.IQ)) {
               bindId = new ID(top.getFirstElementNamed(Protocol.BIND).getFirstElementNamed(Protocol.JID).getText());
            } else {
               closeSocket(true);
               throw new XMLException("The bind request failed.");
            }
            if (isFine) {
               LOG.fine("Bound to "+bindId);
            }
         } else {
            closeSocket(false);
            throw new XMLException("No response from bind request.");
         }

         ItemDestination sessionStanza = outgoing.next();
         Element sessionIQ = constructor.createElement(Protocol.IQ);
         sessionIQ.setAttributeValue("type","set");
         sessionIQ.setAttributeValue("id", "session");
         sessionStanza.send(bindIQ);
         sessionStanza.send(constructor.createElement(Protocol.SESSION));
         sessionStanza.send(constructor.createElementEnd(Protocol.SESSION));
         sessionStanza.send(constructor.createElementEnd(Protocol.IQ));

         iqDest = new DocumentDestination();
         if (incoming.next(iqDest)) {
            LOG.fine("Session established.");
         } else {
            closeSocket(false);
            throw new XMLException("No response from session request.");
         }
      } catch (IOException ex) {
         closeSocket(false);
         throw ex;
      } catch (XMLException ex) {
         closeSocket(true);
         throw ex;
      } finally {
         connecting = false;
         connected = true;
      }
   }
   
   public void disconnect() 
      throws XMLException
   {
      //LOG.info("Disconnecting, socket="+socket+", outgoing="+outgoing);
      if (socket!=null && !socket.isClosed() && outgoing!=null && outgoing.isActive()) {
         //LOG.info("Ending stream.");
         outgoing.end();
      }
      connected = false;
   }
   
   
   protected void closeSocket()
      throws IOException
   {
      closeSocket(true);
   }
   
   protected void closeSocket(boolean disconnect) 
      throws IOException
   {
      if (disconnect) {
         try {
            disconnect();
            if (serverW!=null && socket!=null && !socket.isClosed()) {
               serverW.flush();
            }
         } catch (XMLException ex) {
         }
      }
      if (socket!=null && !socket.isClosed()) {
         socket.setSoTimeout(2000);
         socket.close();
      }
      socket = null;
   }
   
   public void close() 
      throws IOException
   {
      if (socket!=null) {
         closeSocket();
      }
   }

   public void sendPing()
      throws XMLException
   {
      sendPing(null);
   }

   public void sendPing(ID to)
      throws XMLException
   {
      try {
         ItemDestination stanza = outgoing.next();
         if (stanza==null) {
            return;
         }
         Element iq = constructor.createElement(Protocol.IQ);
         if (to!=null) {
            iq.setAttributeValue("to", to.toString());
         } else {
            iq.setAttributeValue("to", bindId.getDomain());
         }
         iq.setAttributeValue("type","get");
         stanza.send(iq);
         stanza.send(constructor.createElement(Protocol.PING));
         stanza.send(constructor.createElementEnd(Protocol.PING));
         stanza.send(constructor.createElementEnd(Protocol.IQ));
      } catch (XMLException ex) {
         // check if socket is bad
         if (ex.getCause() instanceof java.io.IOException) {
            try {
               socket.setSoTimeout(2000);
               socket.close();
            } catch (IOException ioError) {
            }
            socket = null;
         }
         throw ex;
      }

   }
   
   public void sendPresence(boolean available) 
      throws XMLException
   {
      sendPresence(null,available);
   }
   
   public void sendPresence(ID to,boolean available) 
      throws XMLException
   {
      try {
         ItemDestination stanza = outgoing.next();
         if (stanza==null) {
            return;
         }
         Element presence = constructor.createElement(Protocol.PRESENCE);
         if (to!=null) {
            presence.setAttributeValue("to", to.toString());
         }
         if (!available) {
            presence.setAttributeValue("type","unavailable");
         }
         stanza.send(presence);
         stanza.send(constructor.createElementEnd(Protocol.PRESENCE));
      } catch (XMLException ex) {
         // check if socket is bad
         if (ex.getCause() instanceof java.io.IOException) {
            try {
               socket.setSoTimeout(2000);
               socket.close();
            } catch (IOException ioError) {
            }
            socket = null;
         }
         throw ex;
      }
   }
   
   public void sendProbe(ID to) 
      throws XMLException
   {
      try {
         ItemDestination stanza = outgoing.next();
         Element presence = constructor.createElement(Protocol.PRESENCE);
         presence.setAttributeValue("to", to.toString());
         presence.setAttributeValue("type","probe");
         stanza.send(presence);
         stanza.send(constructor.createElementEnd(Protocol.PRESENCE));
      } catch (XMLException ex) {
         // check if socket is bad
         if (ex.getCause() instanceof java.io.IOException) {
            try {
               socket.setSoTimeout(2000);
               socket.close();
            } catch (IOException ioError) {
            }
            socket = null;
         }
         throw ex;
      }
   }
   
   public void startReceiver(final DestinationSource source) {
      receiverRunning = new AtomicBoolean(true);
      receiverThread = new Thread(threadGroup,new Runnable() {
         public void run() {
            while (receiverRunning.get()) {
               if (autoReconnect && (socket==null || !socket.isConnected() || socket.isClosed())) {
                  LOG.fine("Reconnecting...");
                  try {
                     connect(id);
                  } catch (Exception ex) {
                     LOG.log(Level.SEVERE,ex.getMessage(),ex);
                     try {
                        LOG.fine("Waiting "+(reconnectWait/1000)+"s to reconnect.");
                        Thread.currentThread().sleep(reconnectWait);
                     } catch (InterruptedException interrupt) {
                        
                     }
                  }
               }
               if (socket!=null && socket.isConnected() && !socket.isClosed()) {
                  boolean gotStanza = false;
                  boolean isTimeout = false;
                  try {
                     gotStanza =incoming.next(source.getDestination());
                     //LOG.info("gotStanza="+gotStanza);
                  } catch (Exception ex) {
                      gotStanza = false;
                     // hack for timeout being masked by stream parser.  The parse catches the exception
                     // but does not set the cause!
                     long elapsed = (System.currentTimeMillis()-incoming.getLastReceivedAt());
                     //LOG.info("Elapsed="+elapsed);
                     isTimeout = ex.getMessage().indexOf("timed out")>=0;
                     //LOG.info("keepAliveSentAt="+keepAliveSentAt);
                     if (isTimeout && elapsed<idleTimeout) {
                        // we've just timed out because the read timed out.  The keep alives are going through
                        // so we'll fake like we've got a stanza
                        gotStanza = true;
                     }
                     if (!gotStanza && connected) {
                        LOG.log(Level.SEVERE,ex.getMessage(),ex);
                        if (ex.getCause()!=null) {
                           LOG.severe("Cause: "+ex.getCause().getClass().getName());
                        }
                     }
                  }
                  //LOG.info("Got stanza: "+gotStanza+", autoReconnect="+autoReconnect+",socket="+socket+",receiverRunning="+receiverRunning.get());
                  try {
                     if (!gotStanza) {
                        LOG.fine("No stanza recived, shutting down connection...");
                        if (socket!=null) {
                           if (isTimeout) {
                              // since we've timed out, the socket will hang on close
                              socket = null;
                           } else {
                              try {
                                 socket.setSoTimeout(2000);  // we'll wait two seconds for a close
                                 if (socket!=null) {
                                    socket.close();
                                 }
                                 socket = null;
                              } catch (IOException ioError) {
                              }
                           }
                        }
                        if (!autoReconnect) {
                           //LOG.info("Stopping loop...");
                           receiverRunning.set(false);
                        }
                     }
                     //LOG.info("Got stanza: "+gotStanza+", autoReconnect="+autoReconnect+",socket="+socket+",receiverRunning="+receiverRunning.get()+",isConnected="+ClientConnection.this.isConnected());
                  } catch (Exception ex) {
                     LOG.log(Level.SEVERE,"Cannot process no stanza received.",ex);
                  }
               } else if (!autoReconnect) {
                  receiverRunning.set(false);
               }
            }
            LOG.fine("Receiver loop exiting...");
            if (socket!=null && socket.isClosed()) {
               LOG.fine("Socket closed.");
               socket = null;
               outgoing = null;
               serverW = null;
            }
         }
      },"receiver "+id);
      receiverThread.start();
      
   }
   
   public void stopReceiver() {
      if (receiverRunning!=null && receiverRunning.get()) {
         receiverRunning.set(false);
         receiverThread.interrupt();
         try {
            receiverThread.join();
         } catch (InterruptedException ex) {

         }
      }
   }
   
   public boolean isConnecting() {
      return connecting;
   }
   
   public boolean isConnected() {
      boolean value = !connecting && socket!=null && !socket.isClosed() && socket.isConnected();
      if (!value) {
         socket = null;
         outgoing = null;
         serverW = null;
      }
      return value;
   }
   
   public void startKeepAlive() {
      startKeepAlive(60*1000);
   }
   public void startKeepAlive(final int interval) {
      if (keepAliveRunning.get()) {
         return;
      }
      keepAliveRunning.set(true);
      keepAliveThread = new Thread(threadGroup,new Runnable() {
         public void run() {
            while (socket!=null && socket.isConnected() && keepAliveRunning.get()) {
               try {
                  try {
                     Thread.currentThread().sleep(interval);
                  } catch (InterruptedException ex) {
                     // that means we might stop
                  }
                  if (socket!=null && socket.isConnected() && keepAliveRunning.get()) {
                     sendPing();
                     keepAliveSentAt = System.currentTimeMillis();
                     LOG.fine("pingAt="+keepAliveSentAt);
                  }
               } catch (XMLException ex) {
                  LOG.log(Level.SEVERE,ex.getMessage(),ex);
                  keepAliveRunning.set(false);
               }
            }
            LOG.fine("Keep alive exiting...");
            if (socket!=null && socket.isClosed()) {
               socket = null;
               outgoing = null;
               serverW = null;
            }
         }
      },"keep-alive "+id);
      keepAliveThread.start();
   }
   
   public void stopKeepAlive() {
      if (keepAliveRunning!=null && keepAliveRunning.get()) {
         keepAliveRunning.set(false);
         try {
            keepAliveThread.interrupt();
            keepAliveThread.join();
         } catch (InterruptedException ex) {

         }
      }
   }
   
   
   public ItemDestination getStanzaDestination() {
      return isConnected() ? outgoing.next() : null;
   }
   
}
